package com.orion.config;

import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.TopicPartition;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.listener.ConsumerAwareListenerErrorHandler;
import org.springframework.kafka.listener.ListenerExecutionFailedException;
import org.springframework.messaging.Message;

import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @author Administrator
 * @date 2022/2/16
 */
@Configuration
public class KafkaListenerAnnotationErrorHandler {

    /**
     * 无限重试，不采用，仅当记录
     * @return
     */
    @Bean
    public ConsumerAwareListenerErrorHandler listenSeekingErrorHandler() {
        return new ConsumerAwareListenerErrorHandler() {
            @Override
            public Object handleError(Message<?> m, ListenerExecutionFailedException e, Consumer<?, ?> c) {
                List<ConsumerRecord> records = (List<ConsumerRecord>) m.getPayload();
                System.out.println("listen10ErrorHandler...." + e.getMessage());
                if (!records.isEmpty()) {
                    Map<TopicPartition, Long> offsetsToReset = new HashMap<>(16);

                    for (ConsumerRecord<?, ?> record : records) {
                        TopicPartition temp = new TopicPartition(record.topic(), record.partition());

                        Long oldOffset = offsetsToReset.get(temp);
                        Long newOffset = record.offset();

                        if (oldOffset != null && oldOffset > newOffset) {
                            offsetsToReset.put(temp, newOffset);
                        } else {
                            offsetsToReset.put(temp, newOffset);
                        }
                    }
                    System.out.println("offsetsToReset : " + offsetsToReset);
                    offsetsToReset.forEach(c::seek);
                }
                return null;
            }
        };
    }
}
